package com.lvmama.pet.sync.job.impl;

import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.jackson.map.ObjectMapper;

import com.lvmama.comm.bee.service.sync.SyncBaseExecuteService;
import com.lvmama.comm.bee.service.sync.SyncBaseNewService;
import com.lvmama.comm.bee.service.sync.SyncNewException;
import com.lvmama.comm.pet.po.pub.TaskResult;
import com.lvmama.comm.spring.SpringBeanProxy;
import com.lvmama.comm.sync.pojo.SyncTrigger;
import com.lvmama.comm.sync.utils.Constant;
import com.lvmama.pet.sync.job.SyncExecuteJobService;

/**
 * 
 * 任务说明:异构数据同步任务
 * @author zuoxiaoshuai
 * @author pengfei
 *
 */
public class SyncExecuteJobServiceImpl implements SyncExecuteJobService, Serializable {

    private static final long serialVersionUID = 3447543756084801777L;
    
    static Log LOG = LogFactory.getLog(SyncExecuteJobServiceImpl.class);

    static ObjectMapper objectMapper = new ObjectMapper();
    
    public TaskResult execute(Long logId, String parameter) throws Exception {
    	TaskResult  result = new TaskResult();
    	LOG.info(String.format("Task SyncExecuteJobService [%d], parameter [%s], begining ------------------", logId, parameter));
        SyncBaseNewService syncBaseNewService = (SyncBaseNewService) SpringBeanProxy.getBean("syncBaseNewService");
        SyncBaseExecuteService syncExecuteService = (SyncBaseExecuteService) SpringBeanProxy.getBean("syncBaseExecuteService");
        parameter = StringUtils.trimToNull(parameter);
        if (parameter == null) {
        	throw new Exception("参数不能为空");
        }
        Parameter param = null;
        try {
			param = objectMapper.readValue(parameter, Parameter.class);
		} catch (Exception e1) {
			throw new Exception("参数格式错误");
		}
        if (param.getTableNames() == null || param.getTableNames().length == 0) {
        	throw new Exception("TableNames不能为空");
        }
        if (param.getUserNames() == null || param.getUserNames().length == 0) {
        	throw new Exception("UserNames不能为空");
        }
        if (StringUtils.trimToNull(param.getUserOp()) == null) {
        	throw new Exception("UserOp不能为空");
        }
        param.setUserOp(StringUtils.trimToNull(param.getUserOp()).toUpperCase());
        if (!param.getUserOp().equals("IN") && !param.getUserOp().equals("NOT IN")) {
        	throw new Exception("UserOp只能取值IN OR NOT IN");
        }
        param.setTableOp(StringUtils.trimToNull(param.getTableOp()).toUpperCase());
        if (!param.getTableOp().equals("IN") && !param.getTableOp().equals("NOT IN")) {
        	throw new Exception("TableOp只能取值IN OR NOT IN");
        }
        if (param.getCompareValue() != null && param.getFactor() == null) {
        	throw new Exception("CompareValue不为空时Factor不能为空");
        }
        if (param.getCompareValue() == null && param.getFactor() != null) {
        	throw new Exception("Factor不为空时CompareValue不能为空");
        }
        Map<String, Object> condition = new HashMap<String, Object>();
        condition.put("userOp", param.getUserOp());
        condition.put("tableOp", param.getTableOp());
        condition.put("tableNames", param.getTableNames());
        condition.put("userNames", param.getUserNames());
        condition.put("compareValue", param.getCompareValue());
        condition.put("field", param.getField());
        condition.put("factor", param.getFactor());
        condition.put("rows", param.getRows());
        condition.put("batchMark", param.getBatchMark());
        
        List<SyncTrigger> triggers = syncBaseNewService.findTriggerPageListWithTime(condition);
        List<String> sucList = new ArrayList<String>();
        List<String> errorList = new ArrayList<String>();
        List<String> runList = new ArrayList<String>();
        try {
        	//是否需要自动补偿插入一条记录
        	if(StringUtils.isNotEmpty(param.getAutoCreate())&&triggers.size()==0){
            	SyncTrigger syncTrigger = new SyncTrigger();
            	syncTrigger.setUserName(param.getUserNames()[0]);
            	syncTrigger.setTableName(param.getTableNames()[0]);
            	syncTrigger.setTriggerType(param.getTriggerType());
            	syncTrigger.setDataSourceId(param.getDataSourceId());
            	syncTrigger.setStatus("INIT");
            	syncTrigger.setPkValue("99999");
            	syncTrigger.setpLevel(0l);
            	syncBaseNewService.saveSyncTrigger(syncTrigger);
            	LOG.info("auto create SyncTrigger :"+syncTrigger.toString());
            	
            }
        	for (SyncTrigger trigger : triggers) {
        		try {
        			//批量是否异步执行 
        			if(StringUtils.isNotEmpty(param.getAsyncMark())){
        				if(StringUtils.isNotEmpty(trigger.getPkBegin())&&null!=trigger.getPkQuantity()){
                			//手动添加的批量异步执行 先将状态改为执行中  回调修改状态 fail or success
                			syncExecuteService.batchSync(trigger);
            				runList.add(trigger.getTriggerId());
            			}else{
                			syncExecuteService.sync(trigger);
                			sucList.add(trigger.getTriggerId());
            			}
        			}else{
        				executeAutoBatch(trigger, syncExecuteService,syncBaseNewService,sucList,errorList);
        			}

        		} catch (SyncNewException e) {
        			errorList.add(trigger.getTriggerId());
        			LOG.error(e.getMessage(), e);
        		}
        	}
        } finally {
        	if (sucList.size() > 0) {
        		syncBaseNewService.updateTriggersStatus(sucList,   Constant.SYNC_STATUS.SUCCESS.name());
        	}
        	if (errorList.size() > 0) {
        		syncBaseNewService.updateTriggersStatus(errorList, Constant.SYNC_STATUS.FAIL.name());
        	}
        	if (runList.size() > 0) {
        		syncBaseNewService.updateTriggersStatus(runList, Constant.SYNC_STATUS.RUNNING.name());
        	}
        	
        }
        result.setRunStatus(TaskResult.RUN_STATUS.SUCCESS);
        result.setResult("共同步" + (sucList.size() + errorList.size()) + "条，成功"+ sucList.size() + "条，失败" + errorList.size() + "条");
        LOG.info(String.format("Task SyncExecuteJobService [%d], parameter [%s], end ------------------", logId, parameter));
        return result;
    }
    
    public void executeAutoBatch(SyncTrigger trigger,SyncBaseExecuteService syncExecuteService,SyncBaseNewService syncBaseNewService,List<String> sucList,List<String> errorList){
    	try {
    		if(StringUtils.isNotEmpty(trigger.getPkBegin())&&null!=trigger.getPkQuantity()){
    			Long value = 0l;
    	        List<String> runList = new ArrayList<String>();
    	        runList.add(trigger.getTriggerId());
				syncBaseNewService.updateTriggersStatus(runList, Constant.SYNC_STATUS.RUNNING.name());
				Long completeCount=0l;
				Long oriPkValue= Long.parseLong(trigger.getPkValue());
				Long successNum = 0l;
				Long failNum =  0l;
				for(int i=0;i<trigger.getPkQuantity();i++){
						if(i==0){
							value = Long.parseLong(trigger.getPkValue());
						}else{
							value = value+1;
							trigger.setPkValue(String.valueOf(value));
						}
						try {
							syncExecuteService.sync(trigger);
							successNum++;
						} catch (Exception e) {
							//批量执行失败 重新插入一条记录
							SyncTrigger dest =  new SyncTrigger();
							try {
								PropertyUtils.copyProperties(dest, trigger);
								dest.setPkBegin(null);
								dest.setPkQuantity(null);
								dest.setSuccessNum(null);
								dest.setFailNum(null);
								dest.setRemark(e.getMessage());
								syncBaseNewService.saveSyncTrigger(dest);
							}catch (Exception e1) {
							
								LOG.error(e.getMessage(),e1);
							}
							failNum++;
							continue;
						}
						completeCount = (value-oriPkValue)+1;
						//更新syncTrigger completeCount
						syncBaseNewService.updateCompleteCountById(trigger.getTriggerId(), completeCount,successNum,failNum);
				}
				
			}else{
				syncExecuteService.sync(trigger);
			}
			sucList.add(trigger.getTriggerId());
		} catch (SyncNewException e) {
			errorList.add(trigger.getTriggerId());
			throw new SyncNewException(e.getMessage());
		}
    }
    
    static class Parameter implements Serializable {
    	/**
		 * 
		 */
		private static final long serialVersionUID = 1L;

		private Integer compareValue;
    	
    	private Integer factor;
    	
    	private Integer rows;
    	
    	private String[] userNames;
    	
    	private String[] tableNames;
    	
    	private String userOp;
    	
    	private String tableOp;
    	
    	private String field;
    	
    	//是否同步执行批量syncTriger
    	private String batchMark;
    	
    	private String asyncMark;
    	
    	private String autoCreate;
    	
    	private String triggerType;
    	
    	private String dataSourceId;

		public Integer getCompareValue() {
			return compareValue;
		}

		public void setCompareValue(Integer compareValue) {
			this.compareValue = compareValue;
		}

		public Integer getFactor() {
			return factor;
		}

		public void setFactor(Integer factor) {
			this.factor = factor;
		}

		public String[] getUserNames() {
			return userNames;
		}

		public void setUserNames(String[] userNames) {
			this.userNames = userNames;
		}

		public String[] getTableNames() {
			return tableNames;
		}

		public void setTableNames(String[] tableNames) {
			this.tableNames = tableNames;
		}

		public String getUserOp() {
			return userOp;
		}

		public void setUserOp(String userOp) {
			this.userOp = userOp;
		}

		public String getTableOp() {
			return tableOp;
		}

		public void setTableOp(String tableOp) {
			this.tableOp = tableOp;
		}

		public Integer getRows() {
			if (rows == null || rows == 0 || rows > 1000) {
				rows = 1000;
			}
			return rows;
		}

		public void setRows(Integer rows) {
			this.rows = rows;
		}

		public String getField() {
			if ("fieldThree".equals(field)) {
				return field;
			}
			return "pkValue";
		}

		public void setField(String field) {
			this.field = field;
		}

		public String getBatchMark() {
			return batchMark;
		}

		public void setBatchMark(String batchMark) {
			this.batchMark = batchMark;
		}

		public String getAsyncMark() {
			return asyncMark;
		}

		public void setAsyncMark(String asyncMark) {
			this.asyncMark = asyncMark;
		}

		public String getAutoCreate() {
			return autoCreate;
		}

		public void setAutoCreate(String autoCreate) {
			this.autoCreate = autoCreate;
		}

		public String getTriggerType() {
			return triggerType;
		}

		public void setTriggerType(String triggerType) {
			this.triggerType = triggerType;
		}

		public String getDataSourceId() {
			return dataSourceId;
		}

		public void setDataSourceId(String dataSourceId) {
			this.dataSourceId = dataSourceId;
		}
		
		
    }
    
}
